Repository navigation
chore(streaming): re-sync #784 with main after #802, #811, #813 and #814 - #821
Merged
fas89 merged 10 commits intoNov 4, 2026
Merged
Conversation
…ss, not a second lookup (#814) * security(acquisition): the orchestrator guard judges the pinned address, not a second lookup request_guard (#758) is the per-request check that keeps a contract-named deployment.server_url from receiving the connector config, resolved source password included, unless the host is private, loopback or listed in FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST. It re-ran the name check, which resolved the hostname again instead of judging the IP safe_httpx_client's pin hook had already written into request.url. The two lookups can disagree: - DNS rebinding: the pin lookup answers public, the guard's answers private, so an unlisted public address received the config. A listed name could be pointed at instance metadata the same way. - IDNA: socket.getaddrinfo encodes faß.de as IDNA2003 fass.de while httpx resolves IDNA2008 xn--fa-hia.de. Both reproduced end to end against the real KafkaConnectRestClient and pin hook (getaddrinfo faked, transport recorded): on main the request reached the transport for 93.184.216.34; now it is refused before it leaves. The guard now runs the name check as before, then pinned_verdict on request.url.host: link-local is refused always, the URL's name decides the allowlist, the pinned address decides the scope, and the refusal names both ("host 'x', connecting to '93.184.216.34', is public"). VerdictCode, HostVerdict (with pinned), host_verdict and pinned_verdict are copied verbatim from the managed Flink branch (5466a50), where catalog_request_guard already works this way, so the #784 sync merges these hunks cleanly. host_verdict returns no-host for http://[::1 instead of raising ValueError, so a malformed server_url is refused, not a traceback. The unit suite replaces the pin hook with a no-op, so the guard sees the mock orchestrator's name, which does not resolve and is refused. The kafka_connect_mock and airbyte_mock fixtures now list their host, as an operator would for an orchestrator of theirs. The same dial-time shape: richardwooding/ssrfguard's Guard.Control and the datamodel-code-generator 0.63.0 fix for CVE-2026-55391. * chore(types): put the acquisition egress policies under mypy --strict _deployment_host_policy and _secret_policy decide which host a contract-named URL may reach and which variable may be sent there. A silent Any is a silent allow, and strict equality is what makes a misspelt VerdictCode fail CI instead of falling through to the "public" wording. Same entries and comment as the managed Flink branch (5466a50), minus fluid_build.streaming.policy, which is not on main. * docs(changelog): the orchestrator guard judges the pinned address Security entry for the request_guard fix. * fix(acquisition): say the request has no address when nothing was pinned A pinned verdict with no address is not a URL without a host, so the refusal no longer reads "it has no host" there. Review finding. * test(acquisition): drive the guard through the real pin hook The unit suite replaces safe_http's pin hook with a no-op, and the guard tests hand it a stand-in request, so nothing checked the ordering the fix relies on: the pin hook runs first and the guard reads the IP it wrote. Both reviewers showed that moving the pin hook after the guard left the suite green while the faß.de case reached a public address again. The new file is on conftest's real-guard list. It drives the real KafkaConnectRestClient and AirbyteRestClient, with getaddrinfo faked and the transport recording instead of sending. A rebinding name and faß.de are refused with nothing sent, and a private orchestrator is reached at its pinned address. Moving the pin hook last, or restoring the old guard, fails 3 of its 4 tests.
… through a pattern (#811) * security: the ODCS and plugin-extension checks let a trailing newline through a pattern The FLUID fix (schema_validator, #799) left six JSON Schema checks on jsonschema's own pattern keyword, Python's re.search, whose $ also matches before a final newline: - forge/core/artifact_validators.py (fluid validate-artifacts: ODCS, Bitol ODPS) - providers/odcs/validation.py::collect_errors (fluid odcs validate) - providers/odps_standard/validation.py::validate - providers/opds/validator.py (pinned Draft202012Validator) - forge_datamodel/emit/osi_sidecar.py (the copilot's Ossie check) - extension_schemas.py::assemble_proposed_extensions (pinned Draft7Validator) So ODCS refused schema[].id "t 1" but took "t1\n", which an id: | block produces, and the copilot kept an LLM-proposed customScaffold block with use: "ci:basic\n", ref: "main\n" or id: "ci\n". All six now go through schema_validator, and say so with the same line-break hint. schema_validator takes a default dialect for a schema that declares none, passed to jsonschema's validator_for(default=), the way OpenStack Rally uses it. An extension schema is read in the dialect it declares, draft-07 when it declares none (the documented plugin contract), so a 2020-12 schema's unevaluatedProperties: false is enforced instead of ignored. OPDS keeps 2020-12 as its default. patternProperties keys keep re.search. jsonschema matches them in patternProperties, additionalProperties and unevaluatedProperties, and check-jsonschema's override of patternProperties alone lets a key that one keyword matches and another does not through additionalProperties: false unchecked. Only OPDS has key patterns (^[a-z]{2}$, three places). The tripwires now cover every shipped schema, pin that set, and also fail on a $recursiveRef or $dynamicRef, which re-pick the stock class as a nested $schema does. Follows #799. * security: review follow-ups for the vendored-schema end-anchor change - fluid odcs validate cuts each schema message to 120 characters, so for an id like customer_orders_v1 the hint ended "... block keeps and |", which reads as advice to use |. _brief cuts the message ahead of the hint instead; the hint is schema_manager.LINE_BREAK_HINT now, so the command can find it. - schema_validator raises ImportError when jsonschema could not be imported. validate_osi_document, the one unguarded caller, raised NameError: jsonschema instead. validate_opds reads the same JSONSCHEMA_AVAILABLE flag as the other checks; the find_spec probe that found an unimportable jsonschema is gone. - A test walks fluid_build/ and fails on any validator_for(), Draft*Validator() or jsonschema.validate() without cls= outside schema_manager, so a check added later cannot read $ as Python does again. It finds the six sites on the base and seven on the old main. - Tests for the flag branches of collect_errors and the artifact validators, and for OPDS reading an undeclared schema as 2020-12. - CHANGELOG: the extension check shows no message, so it carries no hint. - odps_standard.validate validates against the schema for the document's apiVersion, not only v1.0.0. * fix: a bracket in a value or an unusable plugin schema no longer stops the run Defect-scan follow-ups for the vendored-schema end-anchor change: - fluid odcs validate printed each schema error as Rich markup, so a failing value with "[/]" in it ended the command with "Unexpected error: closing tag '[/]' ... has nothing to close" (exit 2, also on main). The line is printed with markup=False, like the vowl line in the same function. - assemble_proposed_extensions raised when a plugin schema's $schema was not a string: validator_for cannot look it up, where Draft7Validator ignored it. A schema that cannot be applied now drops its block with a redacted warning, as a block that fails its schema is dropped, and the run goes on. This also covers {"type": 5}, which raised on main. - The guard test also catches `from jsonschema import validate`, `import jsonschema as js` with js.validate(...), jsonschema.validators.validate/extend/create, and a cls= that does not come from schema_validator. ODPS passes cls=type(schema_validator(...)) inline so the guard can see it. It parses only files that mention jsonschema. * docs(changelog): keep the two pattern end-anchor entries together The rebase onto main put two new fluid diff --baseline entries between the FLUID-schema entry (#799) and this branch's, which begins "The fix above covered the FLUID schemas only".
* feat(validate,plan): warn about an id the build phase refuses The schema's $defs.identifier admits ids the build phase refuses: a leading digit (1orders) or more than 128 characters. fluid apply in a build mode refuses such an id before it applies anything (#803), but fluid validate and fluid plan said nothing, so the first sign was a failed apply. The two grammars stay different by decision. For a contract with builds, both now warn and name the id, using the rule apply uses (build_runners._ids.validate_build_ids): - validate adds a ValidationResult warning, so plain validate passes and --strict fails (founder's call). It runs with the other 0.7.x contract rules, on a contract file and on a bundle's resolved contract. If the check itself crashes, that is reported as a warning. - plan prints the warning and still writes plan.json, whatever its --mode (founder's call). It runs after the gate plan shares with apply, not inside it, so apply keeps its typed CLIError. IDENT_RULE states the grammar in words, and both the new warning and apply's hint use it. The hint said validate and plan "accept" such an id; it now says they only warn about it. * fix(validate,plan): keep the build-id warning true on a schema-refused id Review follow-ups to the build-id warning: - validate runs its contract rules on a contract the schema refused too, so the warning could say "The contract schema allows more" about an id the schema had just refused (id: 123). The text now only says what the build phase needs and that fluid apply in a build mode refuses the id. - plan prints the warning the way validate prints its warnings: literal, soft_wrap (the terminal wraps, so Rich never breaks a line inside the id) and no highlighter.
#802) * fix(diff): compare column types strictly enough for --fail-on-breaking Now that fluid diff --baseline reads 0.7.x columns, its type comparison decides --fail-on-breaking for those contracts, and several type changes that break consumers passed as widenings or as info. - Fold case for ASCII type names only. Text the comparison cannot order (non-ASCII, or parameters that are not plain integers, such as enum('a','b') or varchar(max)) is its own type, so it no longer reads as unchanged, as "no parameters" or as a whitespace difference. The finding says so when a non-ASCII name is the reason. - A parameter change that is neither a pure widening nor a pure narrowing is breaking (it was info): varchar -> varchar(10), decimal(10,2) -> decimal(12,1), decimal(10,2) -> decimal. - A bigger parameter widens only for types whose parameter measures capacity (lengths, precisions, the decimal family), not for ones that name something, such as geometry(srid). - decimal, numeric and number widen only when the scale and the integer digits (precision - scale) both hold. - A change of base type widens only when the new type declares no parameters, so int -> decimal(5,0) is breaking. - time -> timestamp, float -> decimal/numeric and int -> float leave the widening table. - The type grammar is ASCII only and is matched in linear time. Prior art: the Apache Iceberg table spec ("Schema Evolution": int -> long, float -> double, date -> timestamp from v3, decimal(P,S) -> decimal(P',S) for P' > P) and Delta Lake "Type widening" (decimal(p,s) -> decimal(p+k1,s+k2) for k1 >= k2 >= 0). * fix(diff): report a name the new contract declares twice The comparison indexes exposes, columns and nested fields by name and keeps the last of a repeated name, so an earlier declaration was never compared against the baseline. An expose id, column name or nested field name that the new contract declares more than once is now a breaking change (expose_duplicate_id, column_duplicate_name, nested_field_duplicate_name), in an expose that is new as well as in one both contracts have, and whatever the baseline holds: compare_contracts is idempotent except for a contract that declares a name twice. The declarations are in the finding's "after". * test(diff): diff the schema-invalid fixtures with validate=False fluid diff --baseline validates both contracts since #805, so a fixture that is not a valid contract (the flat-layout and malformed-schema ones) is diffed with validate=False, as the neighbouring tests are.
A plain merge of origin/main (24f1773). Since #815, main is a real ancestor, so git's base (635ea30) is the right one. Resolutions, two conflicts: - fluid_build/build_runners/_deployment_host_policy.py: - #814 already carries #809's `VerdictCode`, `HostVerdict` (with `pinned`), `host_verdict`, `pinned_verdict` and `host_refusal`; their ASTs are identical on both sides. - The conflict was #809's `is_listed`, which main lacks. Kept. - Main's #814 guard (`_host_error`, `_pinned_refusal`, `refuse_unless_host_allowed`, `request_guard`) is unchanged. - pyproject.toml: #814 adds the same two mypy-strict modules and comment as #809. Kept #809's third module, `fluid_build.streaming.policy`. The file now equals the feature branch's. Both sides also changed these files, and git merged them without conflict: - changelog_rules.py: only `diff_columns` changed on both sides: #802's type-change wording, and #784's `_diff_container_children` call. - plan.py: only `run()`: #813's `_warn_unusable_build_ids` runs after the plan gate, ahead of #784's streaming routing. - apply.py and schema_manager.py: no function changed on both sides. In those four files each side's diff is unchanged. The tree differs from main in exactly the feature's 44 files.
#811 added a guard, test_every_validator_is_built_by_schema_validator, which fails on any validator built outside schema_manager.schema_validator. streaming/compiler.py built a plain Draft202012Validator for parameters.flink, so the merged tree failed that guard. Building it through schema_validator keeps Draft 2020-12 as the dialect and makes each pattern end at the end of the value. So a trailing newline in deployment.name or image is now refused; #809's allowlists did not cover those two fields. Two test_fail_closed cases pin it (red before this change).
The feature branch moved during review: #817 (Kafka record shapes declared in FLUID) was squash-merged onto 0b4fc36 as a89966a. Merging it keeps this sync mergeable into the feature branch. The base is 0b4fc36. Resolutions, three hunks in two files: - fluid_build/api/changelog.py, compare_contracts: kept both. #817's `diff_streams` stays after the sovereignty rules, where #817 put it. Main's `diff_duplicates` (#802) sits directly above the per-expose comparison its comment refers to. - fluid_build/streaming/compiler.py imports: kept main's `schema_validator` and #817's `streaming.mapping` names. - fluid_build/streaming/compiler.py, compile_build: kept #817's refusal of topic, format and schema in parameters.flink.source, which now live in properties.stream. Then the config is validated through `schema_validator` (3fa5f4c) in place of #817's raw Draft202012Validator. tests/streaming/test_compiler.py merged without conflict. In those three files each side's diff is unchanged. The tree differs from a89966a in exactly main's changes since 635ea30 plus 3fa5f4c's two files.
… are dotenv-protected (#820) litellm 1.108.0 (released 2026-11-04) reads BESPOKE_API_BASE, LAYA_API_BASE, DATABRICKS_AGENT_API_BASE and MICROSOFT_365_COPILOT_API_BASE (new litellm.provider_list entries), CORALBRICKS_API_BASE and INFERENCEAPIS_API_BASE (api_base_env in llms/openai_like/providers.json), and SCALEDOWN_API_BASE (get_secret_str in llms/scaledown/chat/ transformation.py, "the host in SCALEDOWN_API_BASE"). Each picks the host the provider key goes to, so each joins LITELLM_ENDPOINT_NAMES. tests/test_litellm_endpoint_names.py failed on every CI leg that installs the newest litellm (main and all open PRs); it passes against the 1.108.0 wheel (58 passed, 1 failed without the change).
Brings in #820 (litellm 1.108.0's provider endpoints are dotenv-protected), which fixes the endpoint-names drift test that turned every PR's Test legs red. A clean merge: #820 touches CHANGELOG.md and credentials/litellm_endpoints.py, neither of which the feature changes. The merge adds exactly #820's patch to 00e73e2.
The feature branch moved during review: #819 (refuse Iceberg partition transforms that Flink's DDL cannot create) was squash-merged onto a89966a as 33b7bf1. A clean merge with base a89966a. #819 and this branch's 3fa5f4c both change streaming/compiler.py and tests/streaming/test_compiler.py, in hunks that do not overlap; each side's diff is unchanged in both files.
fas89
pushed a commit
that referenced
this pull request
Nov 4, 2026
Managed Flink to Iceberg streaming preview, in FLUID 0.7.7 (author: Dustlab). - Recursive record types. - Kafka record shapes declared in FLUID (properties.stream and properties.mapping). - Generated FlinkDeployment resources. - Plan, apply, verify and diff integration. - Checkpoint and catalog verification. - fluid generate flink. The operator security gate covers several things: - Allowlists for the kube context, namespace, image, service account and Secrets. - Listed hostnames for every destination the job contacts: the catalog, the token endpoint, S3, the Kafka brokers and the Schema Registry. - NAME@host pairing for catalogAuth. - A pinned-IP request guard. - Refusals: no env: refs in catalog text, an https-only OAuth2 endpoint, no SQL hints, and a contract-SQL allowlist. Hardened through stacked PRs #795, #809, #817, #819, #825 and #826, and syncs #815, #821 and #824. The guide is in forge_docs (advanced/managed-flink-iceberg). The remaining live proof is tracked on Trello.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
#784 went
CONFLICTINGagain after the merges to main since #815: #814, #811, #813 and #802. This PR re-syncs the feature branch with main using real merges, the same method as #815.Merge it with "Create a merge commit", never squash.
What
schema_validator(#811's guard)The tree differs from main (24f1773) in exactly the feature's own files (
635ea30e..a89966a0, 56 files). It differs from the feature branch (a89966a) in exactly main's changes since 635ea30, plus 3fa5f4c's two files.Resolutions
d682b7e (base 635ea30, which is a real ancestor since #815):
build_runners/_deployment_host_policy.py: security(acquisition): the orchestrator guard judges the pinned address, not a second lookup #814 already carries security(streaming): operator allowlists and egress rules for the managed Flink lane (#784 run 2) #809'sVerdictCode,HostVerdict(withpinned),host_verdict,pinned_verdictandhost_refusal. The two sides' versions are AST-identical, so nothing needed reconciling. The only conflict was security(streaming): operator allowlists and egress rules for the managed Flink lane (#784 run 2) #809'sis_listed, which main lacks; it is kept. security(acquisition): the orchestrator guard judges the pinned address, not a second lookup #814's guard (_host_error,_pinned_refusal,refuse_unless_host_allowed,request_guard) comes in unchanged.pyproject.toml: security(acquisition): the orchestrator guard judges the pinned address, not a second lookup #814 adds the same two mypy-strict modules and comment as security(streaming): operator allowlists and egress rules for the managed Flink lane (#784 run 2) #809. security(streaming): operator allowlists and egress rules for the managed Flink lane (#784 run 2) #809's third module,fluid_build.streaming.policy, is kept.changelog_rules.diff_columns: fix(diff): harden the column type comparison behind --fail-on-breaking #802's type-change wording next to feat: add managed Flink to Iceberg streaming preview #784's container recursion.plan.run: feat(validate,plan): warn about an id the build phase refuses #813's build-id warning runs after the plan gate, ahead of feat: add managed Flink to Iceberg streaming preview #784's streaming routing.apply.pyandschema_manager.py: no function was changed by both sides.3fa5f4c. #811 added
test_every_validator_is_built_by_schema_validator, and #784'sstreaming/compiler.pybuilt a rawDraft202012Validator, so the merged tree failed it. The validator now comes fromschema_validator(_CONFIG, default=jsonschema.Draft202012Validator). The dialect is unchanged, and patterns now end at the end of the value. A trailing newline indeployment.nameorimagewas accepted before, and #809's allowlists did not catch it. It is now refused. Two newtest_fail_closedcases cover this; both failed before the fix.00e73e2 (base 0b4fc36):
api/changelog.pycompare_contracts: feat(streaming): declare every Kafka record shape in FLUID, checked before deploy #817'sdiff_streamsstays after the sovereignty rules. Main'sdiff_duplicates(fix(diff): harden the column type comparison behind --fail-on-breaking #802) sits directly above the per-expose comparison its comment refers to.streaming/compiler.py: both imports are kept. feat(streaming): declare every Kafka record shape in FLUID, checked before deploy #817's refusal of topic, format and schema inparameters.flink.sourceruns first, then theschema_validatorfrom 3fa5f4c.In every two-sided file, each side's
-U0delta is byte-identical after the merge.Review
An Opus judge, working in its own worktree, gave PASS on all three commits, with nothing merge-induced at MEDIUM or above:
$-anchored whole-string check.Carried forward:
diff_duplicatesdoes not look atbuilds[].properties.stream.schema. Managed builds already refuse duplicates at validation.streaming/types.pyuses\dwithoutre.ASCII. This predates the merge.Gates (head 00e73e2, local)
ruff(0.15.12 and 0.17.3), the dead-code gate andblack24.10.0 pass.black26.10.1 flags the same 33 files as main.lint-imports(6 + 1 contracts kept), license headers, the product guardrail and isolatedmypy_strict.py(11 files) pass.mypy --strict(7 files) pass.tests/streamingcollects 323 tests.tests/build_runners/test_deployment_host_policy.pyandtest_deployment_host_policy_real_pin_hook.py: 70 passed.python:3.12-slimcontainer exits 0 with the baseline unchanged, on this PR's 26 files and on feat: add managed Flink to Iceberg streaming preview #784's 56 files against main.tests/streaming): 11895 passed. The 2 failures are intest_duckdb_sandbox_workspace_root(macOS/var) and fail the same way on pristine main 24f1773.ci.ymlandflink-iceberg.ymlare dispatched on this branch, because ci.yml'spull_requesttrigger only covers PRs to main.